package 练习11_线程池综合使用案例;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @ClassName: ThreadPoolMonitorHelper
 * @Author: AllenSun
 * @Date: 2025/10/25 04:27
 */
public class ThreadPoolMonitorHelper {
    /**
     * 监测线程池状态并打印日志
     *
     * @param executor 线程池实例
     * @param poolName 线程池名称（用于日志标识）
     */
    public static void monitor(ExecutorService executor, String poolName) {
        if (executor instanceof ThreadPoolExecutor) {
            ThreadPoolExecutor tpe = (ThreadPoolExecutor) executor;
            printThreadPoolStatus(tpe, poolName);
        } else if (executor instanceof ScheduledThreadPoolExecutor) {
            ScheduledThreadPoolExecutor stpe = (ScheduledThreadPoolExecutor) executor;
            printThreadPoolStatus(stpe, poolName);
        } else {
            System.out.println("[WARN] Unsupported executor type: " + executor.getClass().getName());
        }
    }

    /**
     * 监测线程状态并打印日志
     *
     * @param groupName 线程组名称（用于日志标识）
     */
    public static void monitorThreads(String groupName) {
        printThreadGroupStatus(groupName);
    }

    /**
     * 打印线程池状态
     */
    private static void printThreadPoolStatus(ThreadPoolExecutor executor, String poolName) {
        // 1. 基础状态
        String status = String.format(
                "\n[线程池状态] %s\n" +
                        "├── 活动线程: %d\n" +
                        "├── 活动核心线程: %d\n" +
                        "├── 活动非核心线程: %d\n" +
                        "├── 核心线程: %d\n" +
                        "├── 最大线程: %d\n" +
                        "├── 队列大小: %d/%d\n" +
                        "├── 任务统计: %d (完成) + %d (执行中) = %d (总数)\n" +
                        "├── 拒绝次数: %d\n" +
                        "├── 拒绝策略: %s\n" +
                        "└── 存活时间: %d秒",
                poolName,
                executor.getActiveCount(),
                executor.getActiveCount()>executor.getCorePoolSize()?
                        executor.getCorePoolSize():executor.getActiveCount(),
                executor.getActiveCount()>executor.getCorePoolSize()?
                        executor.getActiveCount()-executor.getCorePoolSize():0,
                executor.getCorePoolSize(),
                executor.getMaximumPoolSize(),
                executor.getQueue().size(),
                executor.getQueue().remainingCapacity() + executor.getQueue().size(),
                executor.getCompletedTaskCount(),
                executor.getTaskCount() - executor.getCompletedTaskCount(),
                executor.getTaskCount(),
                getRejectedCount(executor),
                getRejectionPolicyName(executor.getRejectedExecutionHandler()),
                executor.getKeepAliveTime(TimeUnit.SECONDS)
        );

        // 2. 队列详情
        String queueInfo = getQueueInfo(executor.getQueue());

        // 3. 线程状态分布
        String threadStates = getThreadStates(executor);

        // 4. 拒绝策略详情
        String rejectionDetails = getRejectionDetails(executor);

        System.out.println(status + "\n" + queueInfo + "\n" + threadStates + "\n" + rejectionDetails);
    }

    /**
     * 打印线程组状态
     */
    private static void printThreadGroupStatus(String groupName) {
        // 获取所有线程
        Map<Thread, StackTraceElement[]> allThreads = Thread.getAllStackTraces();

        // 按状态分组
        Map<Thread.State, Long> stateCount = allThreads.keySet().stream()
                .collect(Collectors.groupingBy(Thread::getState, Collectors.counting()));

        // 按线程组过滤
        Map<Thread.State, Long> groupStateCount = allThreads.keySet().stream()
                .filter(t -> t.getThreadGroup().getName().contains(groupName))
                .collect(Collectors.groupingBy(Thread::getState, Collectors.counting()));

        // 构建状态字符串
        String states = stateCount.entrySet().stream()
                .map(e -> String.format("%s: %d", e.getKey(), e.getValue()))
                .collect(Collectors.joining(", "));

        String groupStates = groupStateCount.entrySet().stream()
                .map(e -> String.format("%s: %d", e.getKey(), e.getValue()))
                .collect(Collectors.joining(", "));

        System.out.printf(
                "\n[线程状态] %s\n" +
                        "├── 系统线程总数: %d\n" +
                        "├── 系统线程状态: %s\n" +
                        "├── 分组线程总数: %d\n" +
                        "└── 分组线程状态: %s\n",
                groupName,
                allThreads.size(),
                states,
                groupStateCount.values().stream().mapToLong(Long::longValue).sum(),
                groupStates
        );
    }

    /**
     * 获取拒绝策略执行次数
     */
    private static long getRejectedCount(ThreadPoolExecutor executor) {
        try {
            // 通过反射获取拒绝次数（适用于大多数实现）
            Field rejectedField = ThreadPoolExecutor.class.getDeclaredField("rejectedCount");
            rejectedField.setAccessible(true);
            return (long) rejectedField.get(executor);
        } catch (Exception e) {
            // 使用自定义计数器
            return RejectedCounter.getCount(executor);
        }
    }

    /**
     * 获取队列详情
     */
    private static String getQueueInfo(BlockingQueue<Runnable> queue) {
        StringBuilder info = new StringBuilder("[队列详情]\n");

        if (queue instanceof LinkedBlockingQueue) {
            info.append("├── 类型: LinkedBlockingQueue\n");
            info.append("└── 特性: 基于链表，吞吐量高");
        } else if (queue instanceof ArrayBlockingQueue) {
            info.append("├── 类型: ArrayBlockingQueue\n");
            info.append("└── 特性: 基于数组，公平性好");
        } else if (queue instanceof SynchronousQueue) {
            info.append("├── 类型: SynchronousQueue\n");
            info.append("└── 特性: 无缓冲，直接传递");
        } else if (queue instanceof PriorityBlockingQueue) {
            info.append("├── 类型: PriorityBlockingQueue\n");
            info.append("└── 特性: 优先级排序");
        } else {
            info.append("├── 类型: ").append(queue.getClass().getSimpleName()).append("\n");
            info.append("└── 特性: 自定义实现");
        }

        return info.toString();
    }

    /**
     * 获取线程状态分布
     */
    private static String getThreadStates(ThreadPoolExecutor executor) {
        // 获取线程工厂
        ThreadFactory factory = executor.getThreadFactory();
        String threadPrefix;

        // 尝试获取线程名前缀
        if (factory instanceof NamedThreadFactory) {
            threadPrefix = ((NamedThreadFactory) factory).getNamePrefix();
        } else {
            threadPrefix = "Unknown";
        }

        // 获取所有线程
        Map<Thread, StackTraceElement[]> allThreads = Thread.getAllStackTraces();

        // 过滤属于该线程池的线程
        Map<Thread.State, Long> stateCount = allThreads.keySet().stream()
                .filter(t -> t.getName().startsWith(threadPrefix))
                .collect(Collectors.groupingBy(Thread::getState, Collectors.counting()));

        // 构建状态字符串
        return "└── 线程状态: " + stateCount.entrySet().stream()
                .map(e -> String.format("%s=%d", e.getKey(), e.getValue()))
                .collect(Collectors.joining(", "));
    }

    /**
     * 获取拒绝策略名称
     */
    private static String getRejectionPolicyName(RejectedExecutionHandler handler) {
        if (handler instanceof ThreadPoolExecutor.AbortPolicy) {
            return "AbortPolicy (抛出异常)";
        } else if (handler instanceof ThreadPoolExecutor.CallerRunsPolicy) {
            return "CallerRunsPolicy (调用者执行)";
        } else if (handler instanceof ThreadPoolExecutor.DiscardPolicy) {
            return "DiscardPolicy (静默丢弃)";
        } else if (handler instanceof ThreadPoolExecutor.DiscardOldestPolicy) {
            return "DiscardOldestPolicy (丢弃最旧)";
        } else if (handler instanceof RejectedCounter.CustomRejectionHandler) {
            return "CustomRejectionHandler (自定义策略)";
        } else {
            return handler.getClass().getSimpleName();
        }
    }

    /**
     * 获取拒绝策略详情
     */
    private static String getRejectionDetails(ThreadPoolExecutor executor) {
        RejectedExecutionHandler handler = executor.getRejectedExecutionHandler();
        StringBuilder details = new StringBuilder("[拒绝策略详情]\n");

        if (handler instanceof ThreadPoolExecutor.AbortPolicy) {
            details.append("├── 策略: 当任务被拒绝时抛出RejectedExecutionException\n");
            details.append("├── 适用场景: 需要严格保证任务不被丢失的关键业务\n");
            details.append("└── 风险: 可能导致调用方异常中断");
        } else if (handler instanceof ThreadPoolExecutor.CallerRunsPolicy) {
            details.append("├── 策略: 由提交任务的线程直接执行被拒绝的任务\n");
            details.append("├── 适用场景: 不希望任务丢失且可接受调用方阻塞的场景\n");
            details.append("└── 风险: 可能阻塞调用线程，影响整体性能");
        } else if (handler instanceof ThreadPoolExecutor.DiscardPolicy) {
            details.append("├── 策略: 静默丢弃被拒绝的任务\n");
            details.append("├── 适用场景: 可容忍任务丢失的非关键业务\n");
            details.append("└── 风险: 任务丢失无任何提示");
        } else if (handler instanceof ThreadPoolExecutor.DiscardOldestPolicy) {
            details.append("├── 策略: 丢弃队列中最旧的任务，然后重试提交新任务\n");
            details.append("├── 适用场景: 可接受丢失旧任务的新任务优先场景\n");
            details.append("└── 风险: 可能丢失重要任务");
        } else if (handler instanceof RejectedCounter.CustomRejectionHandler) {
            details.append("├── 策略: 自定义拒绝处理逻辑\n");
            details.append("├── 行为: ").append(((RejectedCounter.CustomRejectionHandler) handler).getDescription()).append("\n");
            details.append("└── 处理方式: ").append(((RejectedCounter.CustomRejectionHandler) handler).getAction());
        } else {
            details.append("├── 策略: 未知拒绝策略\n");
            details.append("└── 类型: ").append(handler.getClass().getName());
        }

        return details.toString();
    }

    /**
     * 自定义线程工厂（用于线程命名）
     */
    public static class NamedThreadFactory implements ThreadFactory {
        private final String namePrefix;
        private final AtomicInteger threadCount = new AtomicInteger(1);

        public NamedThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix;
        }

        public String getNamePrefix() {
            return namePrefix;
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, namePrefix + "-" + threadCount.getAndIncrement());
            t.setUncaughtExceptionHandler((thread, ex) ->
                    System.err.println("Uncaught exception in " + thread.getName() + ": " + ex));
            return t;
        }
    }

    /**
     * 拒绝策略计数器
     */
    static class RejectedCounter {
        private static final Map<ExecutorService, AtomicInteger> counters = new ConcurrentHashMap<>();

        public static RejectedExecutionHandler createHandler() {
            return (r, executor) -> {
                counters.computeIfAbsent(executor, k -> new AtomicInteger()).incrementAndGet();
                System.err.println("Task rejected: " + r.toString());
            };
        }

        public static long getCount(ExecutorService executor) {
            AtomicInteger counter = counters.get(executor);
            return counter != null ? counter.get() : 0;
        }

        /**
         * 自定义拒绝策略处理器
         */
        public static class CustomRejectionHandler implements RejectedExecutionHandler {
            private final String description;
            private final String action;

            public CustomRejectionHandler(String description, String action) {
                this.description = description;
                this.action = action;
            }

            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                // 记录日志
                System.err.println("Task rejected: " + r.toString());

                // 增加计数器
                counters.computeIfAbsent(executor, k -> new AtomicInteger()).incrementAndGet();

                // 执行自定义操作
                if ("log".equals(action)) {
                    System.err.println("Logging rejected task: " + r);
                } else if ("requeue".equals(action)) {
                    try {
                        executor.getQueue().put(r);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                }
            }

            public String getDescription() {
                return description;
            }

            public String getAction() {
                return action;
            }
        }
    }
}
